package com.ibeifeng.senior.kafka.producer;

import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * 描述:
 * kafka producer
 *
 * @author jiantao7
 * @create 2018-05-10 18:48
 */
public class Producer extends Thread {
    private final KafkaProducer<Integer, String> producer;
    private final String topic;
    private final Boolean isAsync;

    public Producer(String topic, Boolean isAsync) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaProperties.KAFKA_SERVER_URL + ":" + KafkaProperties.KAFKA_SERVER_PORT);
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "spark_read_kafka_to_hdfs");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.ACKS_CONFIG, "1");
        props.put(ProducerConfig.BATCH_SIZE_CONFIG,"10240");
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG,"655360");
        props.put(ProducerConfig.LINGER_MS_CONFIG,"5");

        producer = new KafkaProducer<>(props);
        this.topic = topic;
        this.isAsync = isAsync;
    }

    @Override
    public void run() {
        int messageNo = 1;
        List<JSONObject> jsonObjects = JavaKafkaProducerMessageSender.readFileAsMessage(topic, "C:\\Users\\Administrator\\Desktop\\ge.json");
        boolean cycle = true;
        /*while (cycle)*/ {
            long startTime = System.currentTimeMillis();
            // Send asynchronously
            if (isAsync) {
                for (int i = 0; i < jsonObjects.size(); i++) {
                    // 输出到控制台，方便进行查看
                    String message = jsonObjects.get(i).toJSONString();
                    System.out.println(message);

                    // 发送数据到kafka
                    producer.send(new ProducerRecord<>(topic,
                            messageNo,
                            message), new DemoCallBack(startTime, messageNo, message));
                    messageNo++;
                }
            } else { // Send synchronously
                try {
                    for (int i = 0; i < jsonObjects.size(); i++) {
                        // 输出到控制台，方便进行查看
                        String message = jsonObjects.get(i).toJSONString();
                        System.out.println(message);

                        // 发送数据到kafka
                        producer.send(new ProducerRecord<>(topic,
                                messageNo,
                                message)).get();
                        System.out.println("Sent message: (" + messageNo + ", " + message + ")");
                        messageNo++;
                    }
                } catch (InterruptedException | ExecutionException e) {
                    e.printStackTrace();
                }
            }
            // 会阻塞保证前面所有的消息都发送完
            producer.close();
            cycle = false;
            System.out.println("producer is stop and messageNo = "+messageNo);
        }
    }
}

class DemoCallBack implements Callback {

    private final long startTime;
    private final int key;
    private final String message;

    public DemoCallBack(long startTime, int key, String message) {
        this.startTime = startTime;
        this.key = key;
        this.message = message;
    }

    /**
     * A callback method the user can implement to provide asynchronous handling of request completion. This method will
     * be called when the record sent to the server has been acknowledged. Exactly one of the arguments will be
     * non-null.
     *
     * @param metadata  The metadata for the record that was sent (i.e. the partition and offset). Null if an error
     *                  occurred.
     * @param exception The exception thrown during processing of this record. Null if no error occurred.
     */
    @Override
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        long elapsedTime = System.currentTimeMillis() - startTime;
        if (metadata != null) {
            System.out.println(
                    "message(" + key + ", " + message + ") sent to partition(" + metadata.partition() +
                            "), " +
                            "offset(" + metadata.offset() + ") in " + elapsedTime + " ms");
        } else {
            exception.printStackTrace();
        }
    }
}